package com.example.springboottest.example.kafka.simple;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * 带事务的生产者
 */
public class TransactionConsumeTransformProduce {
    public static final String brokerList = "localhost:9092";
    public static final String topic = "topic-demo";
//    public static final String topic = "top-create";

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        /**
         * bootstrap.servers ：该参数用来指定生产者客户端连接 Kafka 集群所需的 broker
         * 地址清单，具体的内容格式为 hostl:portl,host2:port2 ，可以设置多个地址，中间以逗号隔开
         */
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);

        properties.put("transactional.id", "transactionalId");// 每台机器的事务ID唯一
        properties.put("enable.idempotence", true);// 设置幂等性
//        properties.put("retries", 2); // 重试次数
//        properties.put("batch.size", 100); // 批量发送大小
//        properties.put("buffer.memory", 33554432); // 缓存大小，根据本机内存大小配置
//        properties.put("linger.ms", 1000); // 发送频率，满足任务一个条件发送
//        properties.put("client.id", "producer-2"); // 发送端id,便于统计
        properties.put("producer.id",-1);
//        properties.put("epoch",-1);


        //自己直生产者客户端参数并创建 KafkaProducer 实例
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        //构建所需妥发送的消息
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, "hello,Kafka!");
        //事务必须初始化
        producer.initTransactions();
        try {
            Thread.sleep(2000);
            producer.beginTransaction();
            producer.send(record);
            producer.commitTransaction();
        } catch (Exception e) {
            producer.abortTransaction();
            e.printStackTrace();
        }
        //关闭生产者客户端示例,提供了一个带超时时间的 close()方法，
        producer.close(1, TimeUnit.MINUTES);
    }
}
